#include "HBaseClient.h"
#include "util.h"
#include <rapidjson/document.h>

#include <utility>

using namespace std;
using namespace rapidjson;
using namespace httplib;

HBaseClient::HBaseClient(string ip, int port) : ip(std::move(ip)), port(port) { }

HBaseClient::~HBaseClient() = default;

bool HBaseClient::Ok()
{
    Client client(ip, port);
    auto resp = client.Get("/");
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

optional<string> HBaseClient::GetClusterVersion()
{
    auto path = "/version/cluster";
    Client client(ip, port);
    auto resp = client.Get(path);
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    return resp->body;
}

optional<list<string>> HBaseClient::GetTables()
{
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Get("/", headers);
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    Document dom;
    dom.Parse(resp->body.c_str());
    auto tab = dom["table"].GetArray();
    list<string> names;
    for (auto i = tab.Begin(); i != tab.End(); i++)
    {
        auto name = (*i)["name"].GetString();
        names.emplace_back(string(name));
    }
    return names;
}

std::optional<TableSchema> HBaseClient::GetTableInfo(const string &tableName)
{
    auto path = stringstream();
    path << "/" << tableName << "/schema";
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Get(path.str().c_str(), headers);
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    auto tableSchema = TableSchema();
    tableSchema.FromString(resp->body);
    return tableSchema;
}

bool HBaseClient::CreateTable(const TableSchema &tableSchema)
{
    auto path = stringstream();
    path << "/" << tableSchema.Name << "/schema";
    auto body = RemoveRedundancyChars(tableSchema.ToString());
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Put(path.str().c_str(), headers, body, "application/json");
    if (!resp)
        return false;
    if (resp->status != 201)
        return false;
    return true;
}

bool HBaseClient::UpdateTable(const TableSchema &tableSchema)
{
    auto path = stringstream();
    path << "/" << tableSchema.Name << "/schema";
    auto body = tableSchema.ToString();
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Post(path.str().c_str(), headers, body, "application/json");
    if (!resp)
        return false;
    if (resp->status != 201)
        return false;
    return true;
}

bool HBaseClient::DeleteTable(const string &tableName)
{
    auto path = stringstream();
    path << "/" << tableName << "/schema";
    Client client(ip, port);
    auto resp = client.Delete(path.str().c_str());
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

std::optional<std::list<std::string>> HBaseClient::GetRows(const string &tableName)
{
    auto path = stringstream();
    path << "/" << tableName << "/*";
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Get(path.str().c_str(), headers);
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    Document dom;
    dom.Parse(resp->body.c_str());
    auto tab = dom["Row"].GetArray();
    list<string> names;
    for (auto i = tab.Begin(); i != tab.End(); i++)
    {
        auto name = (*i)["key"].GetString();
        names.emplace_back(Base64Decode(name));
    }
    return names;
}

std::optional<Row> HBaseClient::GetRow(const std::string& tableName, const std::string& rowKey)
{
    auto path = stringstream();
    path << "/" << tableName << "/" << rowKey;
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Get(path.str().c_str(), headers);
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    auto row = Row();
    row.FromString(resp->body);
    return row;
}

bool HBaseClient::CreateRow(const string &tableName, const Row &row)
{
    if (row.Cells.empty())
        return false;

    auto path = stringstream();
    path << "/" << tableName << "/" << row.Key;
    auto body = RemoveRedundancyChars(row.ToString());
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Put(path.str().c_str(), headers, body, "application/json");
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

bool HBaseClient::UpdateRow(const string &tableName, const Row &row)
{
    if (row.Cells.empty())
        return false;

    auto path = stringstream();
    path << "/" << tableName << "/" << row.Key;
    auto body = RemoveRedundancyChars(row.ToString());
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Post(path.str().c_str(), headers, body, "application/json");
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

bool HBaseClient::DeleteRow(const string &tableName, const std::string& rowKey)
{
    auto path = stringstream();
    path << "/" << tableName << "/" << rowKey;
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Delete(path.str().c_str(), headers);
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

bool HBaseClient::PutCells(const string &tableName, const RowsShell& rows)
{
    if (rows.Rows.empty())
        return false;
    for (const auto& i : rows.Rows)
    {
        if(i.Cells.empty())
            return false;
    }
    auto path = stringstream();
    path << "/" << tableName << "/anything";
    auto body = RemoveRedundancyChars(rows.ToString());
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Put(path.str().c_str(), headers, body, "application/json");
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

bool HBaseClient::DeleteCell(const string &tableName, const std::string& rowKey, const string &column)
{
    auto path = stringstream();
    path << "/" << tableName << "/" << rowKey << "/" << column;
    Client client(ip, port);
    auto resp = client.Delete(path.str().c_str());
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

bool HBaseClient::DeleteCellOfColumnFamily(const string &tableName, const std::string& rowKey, const string &columnFamily)
{
    auto path = stringstream();
    path << "/" << tableName << "/" << rowKey << "/" << columnFamily;
    Client client(ip, port);
    auto resp = client.Delete(path.str().c_str());
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

std::optional<Cell> HBaseClient::GetCell(const string &tableName, const std::string& rowKey, const string &column)
{
    auto path = stringstream();
    path << "/" << tableName << "/" << rowKey << "/" + column;
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Get(path.str().c_str(), headers);
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    auto row = Row();
    row.FromString(resp->body);
    return *(row.Cells.begin());
}

std::optional<Row> HBaseClient::GetCellOfColumnFamily(const string &tableName, const string &rowKey, const string &columnFamily)
{
    auto path = stringstream();
    path << "/" << tableName << "/" << rowKey << "/" + columnFamily;
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Get(path.str().c_str(), headers);
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    auto row = Row();
    row.FromString(resp->body);
    return row;
}




